
#include "Worker.h"
#include "Message.h"
#include "Message_RT.h"
#include "Message_System.h"
#include "Message_Kill_Revive.h"
#include "RouterTable.h"
#include "EntryTT.h"
#include "Mailer.h"
//#include <>semaphore.h>
#include <iostream>
#include <stdlib.h>
#include <string>

using namespace std;

Worker::Worker(int serial_number, int number_of_nodes/*, Mailbox* mailer_mailbox*/)
		: OSThread(), _serial_number(serial_number),
					  _number_of_nodes(number_of_nodes),
					  _mailer_inbox(NULL),
					  _is_RT_changed(true),
					  _is_alive(true),
					  _last_yielded_quanta(-1),
					  _status(THREAD_READY),
					  _got_final_message(false),
					  _killed(false),
					  _system_just_initialized(false),
					  _was_blocked_in_last_run(false)
{
	_neighbours     				= new vector<int>();
	_neighbours_killed				= new vector<int>();
	_active							= true;
	init_RT();
	init_neighbours_RT();
	_inbox							= new Mailbox(WORKER_MAILBOX_SIZE);


	// Initialize quanta related data
	if (pthread_mutex_init(&_inbox_empty_lock, NULL) != 0)
	{
		perror("Worker's inbox lock initialization");
		exit(1);
	}
	if (pthread_cond_init(&_inbox_empty_cond, NULL) != 0)
	{
		perror("Worker's inbox condition variable initialization");
		exit(1);
	}

	if (pthread_mutex_init(&_mailer_inbox_empty_lock, NULL) != 0)
	{
		perror("Worker's inbox lock initialization");
		exit(1);
	}
	if (pthread_cond_init(&_mailer_inbox_empty_cond, NULL) != 0)
	{
		perror("Worker's inbox condition variable initialization");
		exit(1);
	}


    //cout << "Worker " << _serial_number << " created." << endl;



}

Worker::~Worker() {
	delete(_neighbours);
	delete(_neighbours_killed);
	delete(_inbox);
	delete(_routering_table);
	delete(_neighbours_routering_tables);
}

void Worker::init_RT()
{
	_routering_table = new RouterTable(_serial_number, _number_of_nodes);
	//say("RT initialized");
}

void Worker::init_neighbours_RT()
{
	_neighbours_routering_tables = new vector<RouterTable*>();
	for (int i = 0; i < _number_of_nodes+1; ++i) {
		_neighbours_routering_tables->push_back(NULL);
	}
}

void Worker::set_neighbour(int neighbour_serial_number)
{
    _neighbours->push_back(neighbour_serial_number);
}

void Worker::kill_or_revive_neighbour(int neighbour_serial_number, int action)
{
	vector<int> *vector_to_delete_from, *vector_to_add_to;
	if ( action == KILL )
	{
		vector_to_delete_from	= _neighbours;
		vector_to_add_to		= _neighbours_killed;
	}
	else if ( action == REVIVE )
	{
		vector_to_delete_from	= _neighbours_killed;
		vector_to_add_to		= _neighbours;
	}

    vector<int>::iterator iter = vector_to_delete_from->begin();
    while( iter != vector_to_delete_from->end() )
    {
        if (*iter == neighbour_serial_number)
        {
        	// save killed neighbor (in order to revive him later):
        	vector_to_add_to->push_back(neighbour_serial_number);

            // erase killed neighbor from neighbors array:
            iter = vector_to_delete_from->erase( iter );
        }
        else
        {
            ++iter;
        }
    }
}

void Worker::print()
{
	if ( Message::print_allowed )
	{
		cout << "\n************ Worker " << _serial_number << " ************" << endl;
		if ( !_active )
		{
			cout << "       Not Active!" << endl;
		}
		cout << "Phase         : " << getPhase() << endl;
		cout << "Quanta left   : " << getQuanta() << endl;
		cout << "Neighbors     : ";
		for (unsigned int i = 0; i < _neighbours->size(); ++i) {
			cout << _neighbours->at(i) << "  ";
		}
		cout << endl;
		cout << "Inbox         : ";
		_inbox->print();
		cout << "Mailer's inbox: ";
		_mailer_inbox->print();
		_routering_table->print();
		cout << "**********************************" << endl;
	}
}

void Worker::put_in_inbox(Message* message)
{
	// lock inbox
	pthread_mutex_lock(&_mailer_inbox_empty_lock);

	pthread_mutex_lock(&_inbox_empty_lock);


	_inbox->put_message(message);
	cout << "Worker " << _serial_number << " has a new message from " << message->getSource_id() << "!" << endl;
	message->print();
	cout << endl;

	// signal if there is any one waiting
	pthread_cond_signal(&_inbox_empty_cond);
	pthread_mutex_unlock(&_inbox_empty_lock);

	pthread_cond_signal(&_mailer_inbox_empty_cond);
	pthread_mutex_unlock(&_mailer_inbox_empty_lock);
}

bool Worker::is_first_iteration()
{
    return getPhase() == 0;
}

struct receive_arg
{
	Message* 	message;
	bool		first_time;
};

void Worker::run()
{
	while ( true )
	{
		while ( _is_alive ){
			// @inv: when reaching this line in the code, there are NO messages to send.
			//bool neighbours_already_updated = false;
			_status = THREAD_RUNNING;
			if ( is_first_iteration() )
			{
				update_neighbours();
			}
			else
			{
				print();
				// receiving first message:
				receive_arg* a = (receive_arg*)malloc(sizeof(receive_arg));
				Message* m;
				a->message 		= m;
				a->first_time	= true;

				recv((void**)&a);

				cout << "Worker " << _serial_number << "'s inbox emptied" << endl;


				switch ( _last_message_type )
				{
					case MESSAGE_RT:

						// receiving rest of messages:
						a->first_time	= false;
						while ( !_killed  &&  !_inbox->is_empty() )
						{
							recv((void**)&a);
						}

						// update current RT according to messages received, and notify all neighbors:
						if ( update_current_RT() )
							update_neighbours();		// send neighbors my RT
						print();
						break;

					case MESSAGE_SYSTEM:

						if ( _got_final_message )
						{
							_got_final_message = false;
						}
						else
						{
							// 1. forward message to closest neighbor (if needed).
							forward_message_to_closest_neighbour(_message_system_that_has_to_be_forwarded);
						}

						break;

					case MESSAGE_KILL:

						// receiving rest of messages:
						a->first_time	= false;
						while ( !_killed  &&  !_inbox->is_empty() )
						{
							recv((void**)&a);
						}
						// update current RT according to messages received, and notify all neighbors:
						if ( !_killed )
						{
							update_current_RT();
							say("Updating neighbors anyway because it's first time after KILL message");
							update_neighbours();						// send neighbors my RT
						}
						print();
						break;

					case MESSAGE_REVIVE:

						// receiving rest of messages:
						a->first_time	= false;
						while ( !_killed  &&  !_inbox->is_empty() )
						{
							recv((void**)&a);
						}
						// update current RT according to messages received, and notify all neighbors:
						if ( !_killed )
						{
							update_current_RT();
							say("Updating neighbors anyway because it's first time after KILL message");
							update_neighbours();						// send neighbors my RT
						}
						print();
						break;

				} // end of switch
			} // end of if ( first_iteration() )


			if ( _killed )
			{
				pthread_mutex_lock(&_worker_killed_lock);
				while ( _killed )
				{
					say("I'm killed, so waiting...");
					_status = THREAD_BLOCKED;			// this is "fake" blocking, the purpose is for the
														// scheduler not to give a killed node message quanta.
					pthread_cond_wait(&_worker_killed_cond, &_worker_killed_lock);
				}
				pthread_mutex_unlock(&_worker_killed_lock);
				say("I was revived, waking up...");
				say("Resuming... changing status from BLOCKED --> READY");
				_status = THREAD_READY;
				yield();
				cout << "Worker " << _serial_number << " has quanta: " << getQuanta() << endl;
				_status = THREAD_RUNNING;
			}
			else
			{
				yield_if_needed();
			}


		}
		//say("not alive. waiting for orders...");
	}
	say("run() ended");
}

void Worker::yield_if_needed()
{
	if ( _inbox->is_empty()  &&  getQuanta() >= 0 )
	{
		cout << "Worker " << _serial_number << " yielded with quanta " << getQuanta() << endl;
		_last_yielded_quanta = getQuanta();
		_status = THREAD_READY;
		yield();
	}
}

bool Worker::studentSend(void *args)
{
	bool slept_on_inbox_empty = false;
	Message* message = (Message*)args;

	pthread_mutex_lock(&_mailer_inbox_empty_lock);
	_mailer_inbox->print();
	while ( _mailer_inbox->is_full() )
	{
		say("Mailer's inbox is full, so waiting...");
		slept_on_inbox_empty = true;
		_status = THREAD_BLOCKED;
		pthread_cond_wait(&_mailer_inbox_empty_cond, &_mailer_inbox_empty_lock);
	}
	this->_mailer_inbox->put_message(message);
	pthread_mutex_unlock(&_mailer_inbox_empty_lock);


	if ( slept_on_inbox_empty )
	{
		say("Mailer's inbox emptied, waking up...");
		say("Resuming... changing status from BLOCKED --> READY");
		_status = THREAD_READY;
		yield();
		cout << "Worker " << _serial_number << " has quanta: " << getQuanta() << endl;
		_status = THREAD_RUNNING;
	}



	if ( Message_RT * message_RT = dynamic_cast<Message_RT*>(message) )
	{
		cout << message_RT->getOriginal_source_id() << " sent " << message->getTarget_id() << " RT message" << endl;
	}
	else if( Message_System * message_System = dynamic_cast<Message_System*>(message) )
	{
		cout << message_System->getSource_id() << " sent " << message->getCurrent_target_id() << " system message" << endl;
	}
	return true;
}



bool Worker::am_I_the_victim(int victim)
{
    return victim == _serial_number;
}

bool Worker::studentRecv(void **args)
{
	receive_arg* a = (receive_arg*)(*args);
	Message* message					= a->message;
	bool	 receiving_first_message	= a->first_time;


	bool slept_on_inbox_empty = false;

	if ( receiving_first_message )
	{
		// receive first message from worker's inbox:
		pthread_mutex_lock(&_inbox_empty_lock);
		while ( _inbox->is_empty() )
		{
			cout << "Worker " << _serial_number << ": Inbox is empty, so waiting..." << endl;
			slept_on_inbox_empty = true;
			_status = THREAD_BLOCKED;
			pthread_cond_wait(&_inbox_empty_cond, &_inbox_empty_lock);
		}
		pthread_mutex_unlock(&_inbox_empty_lock);

			// After waiting (if needed), the process can resume work
		if ( slept_on_inbox_empty )
		{
			say("Inbox refilled, waking up...");
			say("Resuming... changing status from BLOCKED --> READY");
			_status = THREAD_READY;
			yield();
			cout << "Worker " << _serial_number << " has quanta: " << getQuanta() << endl;
			_status = THREAD_RUNNING;
		}
		// CS:
		// Next part will happen only after quanta is given (OR quanta already exists, and worker didn't block):
		cout << "Worker " << _serial_number << " is downloading mail (first message)..." << endl;
		// After waiting (if needed), the process can resume work

		pthread_mutex_lock(&_inbox_mutex);
		message = this->_inbox->get_next_message();
		pthread_mutex_unlock(&_inbox_mutex);
	}
	else		// not receiving_first_message
	{
		say("is downloading mail (rest of messages)...");
		message = this->_inbox->get_next_message();
	}

	// quanta fixing (if needed):
	if ( slept_on_inbox_empty  &&  message != NULL )
	{
		setQuanta(getQuanta() - 1);
	}

	// for statistics:
	_total_number_of_messages_delivered_to_the_thread_from_the_last_quantum++;

	// actions that worker has to do when a message arrives:
	Message_RT * 			message_RT;
	Message_System * 		message_System;
	Message_Kill_Revive * 	message_kill_revive;

	if ( (message_RT = dynamic_cast<Message_RT*>(message)) )
	{
		_last_message_type = MESSAGE_RT;
		insert_neighbour_RT(message_RT);
		cout << message_RT->getCurrent_target_id() << " received from " << message_RT->getSource_id() << " RT message" << endl;
	}
	else if( (message_System = dynamic_cast<Message_System*>(message)) )
	{
		_last_message_type = MESSAGE_SYSTEM;
		if ( is_die_message(message_System) )
		{
			_routering_table->re_init_table();
			init_neighbours_RT();
			say("good bye");
			_status = THREAD_BLOCKED;
			pthread_exit(NULL);
		}
		else if ( is_message_to_me(message_System) )
		{
			handle_message_to_myself(message_System);
		}
		else
		{
			_message_system_that_has_to_be_forwarded = message_System;	// save it for sending it forward in studentSend()
		}
	}
	else if( (message_kill_revive = dynamic_cast<Message_Kill_Revive*>(message)) )
	{
		int action = message_kill_revive->getAction();
		int victim = message_kill_revive->getWorker_id();

		_system_just_initialized = true;

		switch ( action )
		{
			case KILL:
				_last_message_type = MESSAGE_KILL;
				say("Got message KILL");
				message_kill_revive->print();
				if ( am_I_the_victim(victim) )
				{
					_is_alive = false;
					_killed = true;
				}
				else
				{
					kill_or_revive_neighbour(victim, KILL);
					_routering_table->re_init_table();
					init_neighbours_RT();
					say("after getting KILL message, my state is:");
					print();
				}
				break;

			case REVIVE:
				_last_message_type = MESSAGE_REVIVE;
				say("Got message REVIVE");
				message_kill_revive->print();
				if ( am_I_the_victim(victim) )
				{
					_is_alive = true;
					_killed = false;			// redundant. already done in revive() by the mailer.
					_routering_table->re_init_table();
					init_neighbours_RT();
				}
				else
				{
					kill_or_revive_neighbour(victim, REVIVE);
				}
				say("after getting REVIVE message, my state is:");
				print();
				break;
		}
	}
	return true;
}

void Worker::handle_message_to_myself(Message_System* message_System)
{
	_got_final_message = true;
	cout << "|" << endl << "| Message arrived at its destination!" << endl << "| Worker " << _serial_number << " got \"" << message_System->getText() <<
			"\" from " << message_System->getOriginal_source_id() << endl << "|" << endl;
}

bool Worker::is_message_to_me(Message* message)
{
	return message->getTarget_id() == this->_serial_number;
}

void Worker::insert_neighbour_RT(Message* message)
{
	Message_RT *message_RT;
	if( (message_RT = dynamic_cast<Message_RT*>(message)) )
	{
		RouterTable* rt = message_RT->getRouting_table();
		int source 		= message_RT->getSource_id();

		// insert neighbor's <rt> to the array:
		vector<RouterTable*>::iterator iter = _neighbours_routering_tables->begin();

		// increase iterator to source:
		for (int idiot = 0; idiot < source; ++idiot) {
			++iter;
		}

		_neighbours_routering_tables->erase(iter);
		_neighbours_routering_tables->insert(iter, 1, rt);
	}
	else
	{
		//Beep();
		cout << "Worker " << _serial_number << ": Message is not RT message" << endl;
	}
}


//@pre: _neighbours_routering_tables is updated according to incoming messages.
bool Worker::update_current_RT()
{
	bool someone_updated = false;
	cout << "updating current RT...";
	for (int i = 1; i < _routering_table->size(); ++i) {
		int target_i = _routering_table->get_entry(i)->getTarget();

		int minimum_distance 	= _routering_table->get_entry(target_i)->getDistance();
		int next_hop			= _routering_table->get_entry(target_i)->getNext_hop();

		for (unsigned int j = 1; j < _neighbours_routering_tables->size(); ++j) {
			RouterTable* rt = _neighbours_routering_tables->at(j);
			if ( rt != NULL )
			{
				EntryRT* entry = ((RouterTable*)_neighbours_routering_tables->at(j))->get_entry(target_i);
				if ( entry->getDistance()+1 < minimum_distance )
				{
					someone_updated 	= true;
					minimum_distance 	= entry->getDistance() + 1;
					next_hop			= j;
				}
				else if ( entry->getDistance()+1 == minimum_distance  &&  (int)j < next_hop )
				{
					someone_updated		= true;
					next_hop			= j;
				}
			}
		}

		// update current RT with minimum values:
		_routering_table->get_entry(target_i)->setDistance(minimum_distance);
		_routering_table->get_entry(target_i)->setNext_hop(next_hop);

	}
	if (someone_updated)
	{
		cout << " finished. Table changed." << endl;
	}
	else
	{
		cout << " finished. Table NOT changed." << endl;
	}
	return someone_updated;
}

void Worker::update_neighbours()
{
	for (unsigned int i = 0; i < _neighbours->size(); ++i) {
		int neighbour_id = _neighbours->at(i);
		// lock here
		send(new Message_RT(_serial_number, _serial_number, neighbour_id, neighbour_id, _routering_table));
	}
	say("neighbors updated.");
}

void Worker::forward_message_to_closest_neighbour(Message_System* message_system)
{
	int neighbour_id = get_closest_neighbour_that_is_not(message_system->getSource_id());
	cout << "Worker " << _serial_number << ": is forwarding message to " << neighbour_id << endl;
	message_system->setSource_id(_serial_number);
	message_system->setCurrent_target_id(neighbour_id);
	send(message_system);
}

int Worker::get_closest_neighbour_that_is_not(int source_worker_id)
{
	int final_target = _message_system_that_has_to_be_forwarded->getTarget_id();
	if ( is_neighbour(final_target) )
	{
		return final_target;
	}


	int closest_neighbour_id = 0;			// closest neighbour to final target, not to me!
	int smallest_distance 	 = 999999;
	for (unsigned int i = 0; i < _neighbours->size(); ++i) {

		int neighbour_id 		= _neighbours->at(i);
		int neighbour_distance 	= _routering_table->get_entry(neighbour_id)->getDistance();
		//int neighbour_distance 	= ((RouterTable*)_neighbours_routering_tables->at(neighbour_id))->get_entry(final_target)->getDistance();

		if (neighbour_id != source_worker_id)
		{
			if ( neighbour_distance  <  smallest_distance )
			{
				smallest_distance 		= neighbour_distance;
				closest_neighbour_id	= neighbour_id;
			}
		}
	}
	return closest_neighbour_id;
}

void Worker::say(string text)
{
	if ( Message::print_allowed )
	{
		cout << "Worker " << _serial_number << ": " << text << endl;
	}
}

bool Worker::is_neighbour(int worker_id)
{
	for (unsigned int i = 0; i < _neighbours->size(); ++i) {
		if ( _neighbours->at(i) == worker_id )
			return true;
	}
	return false;
}

bool Worker::is_die_message(Message_System* message_System)
{
	return ( message_System->getText().compare("die") ) == 0;
}
